package top.lyunk.demo.flink.job1;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * <pre>
 * 说明：
 * </pre>
 *
 * @author lyk
 * @date 2022/9/6
 */
public class job3 {
    public static void main(String[] args) throws Exception {
        /*
        1. 获取一个执行环境
         */
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        /*
        2. 加载/创建初始数据
         */
        ParameterTool parameterTool = ParameterTool.fromArgs(args);
        DataStreamSource<String> input = env.socketTextStream(parameterTool.get("hostname", "127.0.0.1"), parameterTool.getInt("port", 9999));
        // DataStreamSource<String> input = env.fromElements("aaa,bbb,ccc", "ddd,aaa");

        /*
        3. 指定数据的相关转换
         */
        SingleOutputStreamOperator<WordCount> result = input.flatMap(new WordCountFlatMapFunc());
        // 按key分组
        // result.keyBy(value -> value.f0)
        SingleOutputStreamOperator<WordCount2> dataStream = result.keyBy(WordCount::getWord)
                // .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .sum("count")
                .map(new MapFunc())
                ;


        dataStream.print();
        env.execute("Word Count");
    }

    public static class WordCountFlatMapFunc implements FlatMapFunction<String, WordCount>{
        @Override
        public void flatMap(String in, Collector<WordCount> out) throws Exception {
            for (String word: in.split(",")) {
                out.collect(new WordCount(word, 1));
            }
        }
    }

    public static class MapFunc implements MapFunction<WordCount, WordCount2>{

        @Override
        public WordCount2 map(WordCount value) throws Exception {
            return new WordCount2(value.word, value.count, value.word + ":" + value.count);
        }
    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class WordCount {
        private String word;
        private Integer count;
    }
    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class WordCount2 {
        private String word;
        private Integer count;
        private String str;
    }
}
